package com.ylkj.redis;

import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.hwl.pool.object.BaseCommonPool;
import com.hwl.pool.object.IPooledObjectFactory;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: wengzx
 * @Date: 2025/4/28
 * @Desc:
 */
@RequiredArgsConstructor
@Slf4j
public class RedisSink extends RichSinkFunction<String> {
    private final RedisConfig redisConfig;
    private final String redisKey;
    private Jedis jedis;
    private BaseCommonPool<JSONObject> jsonObjectBaseCommonPool;
    private static volatile Map<String, String> caches = new ConcurrentHashMap<>();

    // 使用 ReentrantLock 确保线程安全
    private final ConcurrentHashMap<String, ReentrantLock> locks = new ConcurrentHashMap<>();
    @Override
    public void open(Configuration parameters) throws Exception {
        log.info("配置信息：{}", redisConfig);
        jedis = redisConfig.getJedis();
        jsonObjectBaseCommonPool = new BaseCommonPool<>(new IPooledObjectFactory<JSONObject>() {
            @Override
            public JSONObject makeObject() {
                return new JSONObject(5);
            }

            @Override
            public void resetObject(JSONObject jsonObject) {
                jsonObject.clear();
            }
        });
    }
    @Override
    public void invoke(String value, Context context) throws Exception {
        if(StringUtils.isBlank(value)){
            log.warn("上报的消息体数据为空{}", value);
            return;
        }
        JSONObject inputObject = JSON.parseObject(value);
        //提取必要的参数
        String deviceCode = inputObject.getString("deviceCode");
        String onlineStatus = inputObject.getString("onlineStatus");
        String reportTime = inputObject.getString("reportTime");
        if (StringUtils.isBlank(deviceCode) || StringUtils.isBlank(reportTime) || StringUtils.isBlank(onlineStatus)) {
            log.warn("必要字段缺失，跳过该记录。");
            return ;
        }
        // 获取针对 deviceCode 的锁，确保线程安全
        ReentrantLock lock = locks.computeIfAbsent(deviceCode, k -> new ReentrantLock());
        lock.lock();
        try {
            // 从缓存或 Redis 中获取现有数据
            String historyDate = caches.get(deviceCode);
            // 如果当前数据在redis中不存在或当前数据的事件时间晚于旧数据时间，则更新或插入到redis
            if (compareDates(historyDate, reportTime) <= 0) {
                JSONObject agentData = createReportData(deviceCode, reportTime, onlineStatus);
                String targetData = agentData.toString();
                jsonObjectBaseCommonPool.returnObject(agentData);
                // 更新缓存
                caches.put(deviceCode, reportTime);

                // 写入Redis
                while (true) {
                    try {
                        jedis.hset(redisKey, deviceCode, targetData);
                        break;
                    } catch (Throwable e) {
                        ThreadUtil.sleep(200);
                        jedis = redisConfig.getJedis();
                        log.error("redis hset failed...");
                        log.error(e.getMessage(), e);
                    }
                }

                if (log.isDebugEnabled()) {
                    log.debug("{} {}更新，比较时间 {} {} value:{}", redisKey, deviceCode, historyDate, reportTime, targetData);
                }
            }
        }finally {
            // 释放锁
            lock.unlock();
            // 清理无用的锁，避免内存泄漏
            locks.remove(deviceCode, lock);
        }
    }

    private JSONObject createReportData(String deviceCode, String reportTime, String onlineStatus) {
        JSONObject agentData = jsonObjectBaseCommonPool.borrowObject();
        agentData.put("DEVICE_CODE", deviceCode);
        agentData.put("ONLINE_STATUS", onlineStatus == null ? " " : onlineStatus);
        agentData.put("UPDATE_TIME", reportTime);
        return agentData;
    }


    // 比较日期时间的大小
    private int compareDates(String date1, String date2) {
        if (date1 == null || date2 == null) {
            return -1;
        }
        if (date1.isEmpty() || date2.isEmpty()) {
            return -1;
        }
        return date1.compareTo(date2);
    }

    @Override
    public void close() throws Exception {
        if (jedis != null) {
            jedis.close();
        }
    }
}
